package com.atguigu.dga.governance.service.impl;

import com.atguigu.dga.ds.bean.TDsTaskDefinition;
import com.atguigu.dga.ds.bean.TDsTaskInstance;
import com.atguigu.dga.ds.service.TDsTaskDefinitionService;
import com.atguigu.dga.ds.service.TDsTaskInstanceService;
import com.atguigu.dga.governance.assess.Assessor;
import com.atguigu.dga.governance.bean.AssessParam;
import com.atguigu.dga.governance.bean.GovernanceAssessDetail;
import com.atguigu.dga.governance.bean.GovernanceMetric;
import com.atguigu.dga.governance.mapper.GovernanceAssessDetailMapper;
import com.atguigu.dga.governance.service.GovernanceAssessDetailService;
import com.atguigu.dga.governance.service.GovernanceMetricService;
import com.atguigu.dga.meta.bean.TableMetaInfo;
import com.atguigu.dga.meta.constant.CodeConst;
import com.atguigu.dga.meta.service.TableMetaInfoService;
import com.atguigu.dga.util.SpringBeanProvider;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * <p>
 * 治理考评结果明细 服务实现类
 * </p>
 *
 * @author zhangchen
 * @since 2023-11-21
 */
@Service
@DS("dga")
public class GovernanceAssessDetailServiceImpl extends ServiceImpl<GovernanceAssessDetailMapper, GovernanceAssessDetail> implements GovernanceAssessDetailService {

    //1  提取要考评的元数据和辅助信息
    //2  提取要考评的指标
    //3  对每张表 每个指标进行考评 -->考评结果明细 list
    //4  把考评结果明细保存

    @Autowired
    TableMetaInfoService tableMetaInfoService;

    @Autowired
    GovernanceMetricService governanceMetricService;

    @Autowired
    SpringBeanProvider springBeanProvider;

    @Autowired
    TDsTaskDefinitionService tDsTaskDefinitionService;

    @Autowired
    TDsTaskInstanceService tDsTaskInstanceService;

    ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(10, 20, 60L, TimeUnit.SECONDS, new LinkedBlockingDeque<>(10000));


    public void  mainAssess(String assessDate){

        // 0 清理今天已经做过的考评
        remove(new QueryWrapper<GovernanceAssessDetail>().eq("assess_date",assessDate));

        //1  提取要考评的元数据和辅助信息
        List<TableMetaInfo>  tableMetaInfoList=tableMetaInfoService.getTableMetaInfoWithExtraList(assessDate);
        Map<String, TableMetaInfo> tableMetaInfoMap = tableMetaInfoList.stream().collect(Collectors.toMap(tableMetaInfo -> tableMetaInfo.getSchemaName() + "." + tableMetaInfo.getTableName(), tableMetaInfo -> tableMetaInfo));

        System.out.println("tableMetaInfoList = " + tableMetaInfoList);

        //2  提取要考评的指标
        List<GovernanceMetric> governanceMetricList = governanceMetricService.list(new QueryWrapper<GovernanceMetric>().eq("is_disabled", "0"));
        System.out.println("governanceMetricList = " + governanceMetricList);

        //2.5   准备ds的任务定义和任务实例
        //实例
        Map<String, List<TDsTaskInstance>> taskInstanceListMap = tDsTaskInstanceService.getTaskInstanceList(assessDate);

        //定义
        //把当天运行最后一次实例对应的任务定义取出来  map<表名,任务定义>

        Map<String, TDsTaskDefinition>  taskDefinitionMap=tDsTaskDefinitionService.getTaskDefinitionMap(assessDate);


        //现货列表
        List<GovernanceAssessDetail> governanceAssessDetailList=new ArrayList<>(tableMetaInfoList.size()*governanceMetricList.size());

        //期货列表
        List<CompletableFuture<GovernanceAssessDetail>> futureList=new ArrayList<>(tableMetaInfoList.size()*governanceMetricList.size());

        //3  对每张表 每个指标进行考评 -->考评结果明细 list
        long startTs = System.currentTimeMillis();
        for (TableMetaInfo tableMetaInfo : tableMetaInfoList) {
            for (GovernanceMetric governanceMetric : governanceMetricList) {


                //张表 每个指标
//                if(governanceMetric.getMetricCode().equals("TEC_OWNER")){
//                    //调用某个考评类某个方法
//                    assessor=new TecOwnerAssessor();
//                } else if (governanceMetric.getMetricCode().equals("BUSI_OWNER")) {
//                    assessor=new BusiOwnerAssessor();
//                }
                String metricCode = governanceMetric.getMetricCode();
                Assessor assessor = springBeanProvider.getBean(metricCode, Assessor.class);

                //封装考评参数
                AssessParam assessParam = new AssessParam();
                assessParam.setAssessDate(assessDate);
                assessParam.setTableMetaInfo(tableMetaInfo);
                assessParam.setGovernanceMetric(governanceMetric);
                assessParam.setAllTableMetaInfoList(tableMetaInfoList);
                assessParam.setAllTableMetaInfoMap(tableMetaInfoMap);

                //获得当前表的当天的任务列表
                List<TDsTaskInstance> tDsTaskInstanceList = taskInstanceListMap.get(tableMetaInfo.getSchemaName() + "." + tableMetaInfo.getTableName());
                assessParam.setTDsTaskInstanceList(tDsTaskInstanceList);

                TDsTaskDefinition tDsTaskDefinition = taskDefinitionMap.get(tableMetaInfo.getSchemaName() + "." + tableMetaInfo.getTableName());
                assessParam.setTDsTaskDefinition(tDsTaskDefinition);

//                AssessParam assessParam1 = AssessParam.builder().assessDate(assessDate)
//                        .allTableMetaInfoMap(tableMetaInfoMap)
//                        .governanceMetric(governanceMetric)
//                        .allTableMetaInfoList(tableMetaInfoList)
//                        .tDsTaskDefinition(tDsTaskDefinition)
//                        .tDsTaskInstanceList(tDsTaskInstanceList)
//                        .tableMetaInfo(tableMetaInfo)
//                        .build();

                if(tableMetaInfo.getTableMetaInfoExtra().getDwLevel().equals(CodeConst.DW_LEVEL_DWD)){
                    System.out.println(1111);
                }
                //执行考评  /////////////需要异步处理///////////

                CompletableFuture<GovernanceAssessDetail> future = CompletableFuture.supplyAsync(() -> {

                    GovernanceAssessDetail governanceAssessDetail = assessor.metricAssess(assessParam);
                    return governanceAssessDetail;

                },threadPoolExecutor);

                // GovernanceAssessDetail governanceAssessDetail = assessor.metricAssess(assessParam);
                ///////////////////////////需要异步处理//////////
               // governanceAssessDetailList.add(governanceAssessDetail);
                futureList.add(future);
            }
        }
        //期货变为现货
         governanceAssessDetailList = futureList.stream().map(future -> future.join()).collect(Collectors.toList());

        long endTs = System.currentTimeMillis();
        System.out.println("考评总耗时 ： " +( endTs-startTs) +"毫秒数");

        //4  把考评结果明细保存
       // System.out.println("governanceAssessDetailList = " + governanceAssessDetailList);

        saveOrUpdateBatch(governanceAssessDetailList);

    }



}
